package com.shujia.seach

import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.table.api.EnvironmentSettings
import org.apache.flink.table.api.bridge.scala._

object CarsToES {
  def main(args: Array[String]): Unit = {

    val configuration = new Configuration()
    configuration.setString("akka.ask.timeout", "1000 s")

    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.createLocalEnvironment(1, configuration)

    /**
      * 创建flink sql环境
      *
      */

    val bsSettings: EnvironmentSettings = EnvironmentSettings
      .newInstance()
      .useBlinkPlanner() //使用blink计划器
      .inStreamingMode() //流模式
      .build()

    /**
      * 创建flink sql 环境
      *
      */

    val bsTableEnv: StreamTableEnvironment = StreamTableEnvironment.create(env, bsSettings)


    /**
      * 1、使用flink sql 读取kafka中的过车数据
      *
      * {"car":"皖AAJQ86",
      * "city_code":"340100",
      * "county_code":"340104",
      * "card":117160031851010,
      * "camera_id":"01230","
      * orientation":"东",
      * "road_id":35727674,
      * "time":1614731544,
      * "speed":89.99}
      */

    bsTableEnv.executeSql(
      """
        |CREATE TABLE cars_kafka (
        | car String,
        | city_code String,
        | county_code String,
        | card BIGINT,
        | camera_id String,
        | orientation String,
        | road_id BIGINT,
        | `time` BIGINT,
        | speed Double
        |) WITH (
        | 'connector' = 'kafka',
        | 'topic' = 'cars',
        | 'properties.bootstrap.servers' = 'master:9092',
        | 'properties.group.id' = 'testGroup',
        | 'format' = 'json',
        | 'scan.startup.mode' = 'earliest-offset'
        |)
        |
      """.stripMargin)


    /**
      * 读取hbase中卡口维表
      *
      */

    bsTableEnv.executeSql(
      """
        |CREATE TABLE card_hbase (
        | card STRING,
        | info ROW<lon DOUBLE,lat DOUBLE>,
        | PRIMARY KEY (card) NOT ENFORCED
        |) WITH (
        | 'connector' = 'hbase-1.4',
        | 'table-name' = 'card',
        | 'zookeeper.quorum' = 'master:2181'
        |)
        |
      """.stripMargin)


    /**
      * 创建一张将数据写入es的表
      *
      * 需要先在es中创建索引库
      * PUT /cars/
      * {
      * "mappings": {
      * "properties": {
      * "ts": {
      * "type": "date",
      * "format": "yyyy-MM-dd HH:mm:ss"
      * },
      * "location": {
      * "type": "geo_point"
      * }
      * }
      * }
      * }
      *
      */

    bsTableEnv.executeSql(
      """
        |CREATE TABLE cars_es (
        | car String,
        | city_code String,
        | county_code String,
        | card BIGINT,
        | camera_id String,
        | orientation String,
        | road_id BIGINT,
        | `time` BIGINT,
        | speed Double,
        | location ARRAY<DOUBLE>,
        |  ts  String,
        |  PRIMARY KEY (car,`time`) NOT ENFORCED
        |) WITH (
        |  'connector' = 'elasticsearch-7',
        |  'hosts' = 'http://master:9200',
        |  'index' = 'cars'
        |)
        |
      """.stripMargin)

    bsTableEnv.executeSql(
      """
        |
        |insert into cars_es
        |
        |select
        | a.car,
        | a.city_code,
        | a.county_code ,
        | a.card,
        | a.camera_id ,
        | a.orientation ,
        | a.road_id ,
        | a.`time` ,
        | a.speed,
        | array[b.info.lon,b.info.lat] as location,
        | FROM_UNIXTIME(a.`time` ) as ts
        |from
        |cars_kafka as a
        |join
        |card_hbase as b
        |on a.card=cast(b.card as bigint)
        |
      """.stripMargin)

  }
}
